package net.wzw.mq.producer;

import lombok.extern.slf4j.Slf4j;
import net.wzw.constant.RocketMQConstant;
import net.wzw.exception.BizException;
import net.wzw.mq.RocketMQProperties;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * @author wzw
 * @date 2025/4/21
 * @Description 通用生产者后续根据情况更换生产者组
 */
@Slf4j
@Component
public class DefaultProducer {

    /**
     * 事务生产者客户端
     */
    private final TransactionMQProducer producer;

    @Autowired
    public DefaultProducer(RocketMQProperties rocketMQProperties) {
          //初始化事务生产者客户端,设置对应的生产者组
        producer = new TransactionMQProducer(RocketMQConstant.FOOD_DEFAULT_PRODUCER_GROUP);
          //设置nameserver
        producer.setNamesrvAddr(rocketMQProperties.getNameServer());
        //启动生产者服务
        start();

    }
    /**
     * 启动rocketmq 生产者服务
     * 该对象在使用之前必须要调用一次，只能初始化一次
     *
     * */
    public void start(){
        try {
            this.producer.start();

        }catch (MQClientException e){

            log.error("rocketmq producer start error",e);

        }

    }

    /**
     * 关闭rocketmq 生产者
     * **/
    public void shutdown(){

        this.producer.shutdown();
    }

    /**
     * 发送单条消息
     * **/
    public void sendMessage(String topic,String message,String type){

        sendMessage(topic,message,-1,type);

    }

    /**
     * Description: 发送消息
     * @param topic 主题 message 消息 delayTimeLevel 延迟等级 type 类型
     * @Return: {@link null}
     * @Author: Administrator
     * @Date: 2025/4/21 23:41
     */
    public void sendMessage(String topic,String message,Integer delayTimeLevel,String type){

        Message msg = new Message(topic,message.getBytes(StandardCharsets.UTF_8));
        try {
            if(delayTimeLevel >0 ){

                msg.setDelayTimeLevel(delayTimeLevel);

            }

             SendResult send =  producer.send(msg);

             if(SendStatus.SEND_OK == send.getSendStatus()){

                 log.info("发送MQ消息成功，type:{},message:{}",type,message);

             }else {

                 throw new BizException(send.getSendStatus().toString());

             }

        }catch (Exception e){
            log.error("发送MQ消息失败：{}",e);
            throw  new BizException("发送 MQ 消息失败");
        }

    }

    /**
     * 批量发送消息
     * topic 主题
     * message 多个消息
     * type 类型
     * **/
    public void sendMessages(String topic, List<String> messages, String type){

        sendMessages(topic,messages,-1,type);

    }

    /**
     * 批量发送消息
     * topic 主题
     * message 多个消息
     * type 类型
     * **/
    public void sendMessages(String topic,List<String> messages,Integer delayTimeLevel,String type){

        List<Message> list = new ArrayList<>();

        for(String message: messages){

            Message msg = new Message(topic,message.getBytes(StandardCharsets.UTF_8));

            if(delayTimeLevel>0){

                msg.setDelayTimeLevel(delayTimeLevel);

            }

            list.add(msg);

        }
        try {

            SendResult send =  producer.send(list);

            if(SendStatus.SEND_OK == send.getSendStatus()){

                log.info("发送MQ消息成功,type:{}",type);

            }
        }catch (Exception e){

            log.error("发送MQ消息失败:",e);
            throw  new BizException("发送MQ消息失败");

        }

    }

    /**
     * 获取事务生产者
     * **/
    public TransactionMQProducer getProducer(){

        return  producer;
    }


}
